package com.gxy.learn.thread.example;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.LinkedList;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @description: 用Java模拟阻塞队列
 * <p>
 * 面试题:写一个固定容量同步容器， 拥有put filget方法，以及getCount方法。
 * 能够支持2个生产者线程以及10个消费者线程的阻塞调用
 * 使用wait notify来实现
 * @author: gaoxueyong
 * @Date: 2022/11/10 18:57
 */
@Slf4j
public class JavaBlockQueenTwoTest<T> {

    private final LinkedList<T> queenList = new LinkedList<>();
    private static final int MAX = 10;
    private static int count = 0;
    private static int MAXCOUNT = 0;

    private Lock lock = new ReentrantLock();
    private Condition producer = lock.newCondition();
    private Condition customer = lock.newCondition();

    /**
     * 生产者投递消息
     *
     * @return void
     * @Author gaoxueyong
     * @Date 7:04 下午 2022/11/10
     * @Param [t]
     **/
    public void put(T t) {
        lock.lock();
        try {
            while (queenList.size() == MAX) {
                try {
                    log.warn("生产者阻塞中。。。");
                    producer.await();
                } catch (Exception e) {
                    log.error("err ", e);
                }
            }
            queenList.add(t);
            ++count;
            ++MAXCOUNT;
//            唤醒所有消费者
            customer.signalAll();
        } catch (Exception e) {
            log.error("err", e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 消费者消费消息
     *
     * @return void
     * @Author gaoxueyong
     * @Date 7:05 下午 2022/11/10
     * @Param []
     **/
    public T get() {
        T t = null;
        try {
            lock.lock();
            while (queenList.size() == 0) {
                try {
                    customer.await();
                } catch (Exception e) {
                    log.error("err ", e);
                }
            }
            t = queenList.removeFirst();
            count--;
            //        通知生产者
            producer.signalAll();
            return t;
        } catch (Exception e) {
            log.error("err", e);
        } finally {
            lock.unlock();
        }
        return t;
    }


    /**
     * 生成32位随机码
     *
     * @return
     */
    public synchronized String random32NumStr() {
        String[] charArray = new String[]{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", "U", "V", "W", "X", "Y", "Z"};
        int maxLen = 32;
        StringBuffer stringBuffer = new StringBuffer();
        for (int i = 0; i < maxLen; i++) {
            stringBuffer.append(charArray[Double.valueOf(Math.floor(Math.random() * maxLen)).intValue()]);
        }
        return stringBuffer.toString();
    }

    public static void main(String[] args) {
        JavaBlockQueenTwoTest<String> r = new JavaBlockQueenTwoTest<>();
//        new Thread(() -> {
//            while (true) {
//                r.put(r.random32NumStr());
//                log.info("共向队列中放入数据[{}]个",MAXCOUNT);
//            }
//        }).start();
//        new Thread(() -> {
//            while (true) {
//                log.info("消费者消费{}，队列还剩余{}个未消费",r.get(),count);
//            }
//        }).start();

        new Thread(() -> {
            ThreadFactory consumptionThreadFactory = new ThreadFactoryBuilder().setNameFormat("生产者线程").build();
            /**
             * maximumPoolSize 不应该超过线程池最小连接数 （spring.datasource.druid.min-idle） 否则会报 get connection timeout retry : 1 异常
             */
            ExecutorService consumptionExecutor = new ThreadPoolExecutor(2, 5,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(10), consumptionThreadFactory, new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    if (!executor.isShutdown()) {
                        try {
                            executor.getQueue().put(r);
                            log.warn("生产者队列阻塞中。。。");
                        } catch (InterruptedException e) {
                            // should not be interrupted
                        }
                    }
                }
            });
            while (true) {
                consumptionExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        r.put(r.random32NumStr());
                        log.info("共向队列中放入数据[{}]个", MAXCOUNT);
                    }
                });
            }
        }).start();

        new Thread(() -> {
            ThreadFactory consumptionThreadFactory = new ThreadFactoryBuilder().setNameFormat("消费者线程").build();
            /**
             * maximumPoolSize 不应该超过线程池最小连接数 （spring.datasource.druid.min-idle） 否则会报 get connection timeout retry : 1 异常
             */
            ExecutorService consumptionExecutor = new ThreadPoolExecutor(2, 5,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(10), consumptionThreadFactory, new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    if (!executor.isShutdown()) {
                        try {
                            executor.getQueue().put(r);
                            log.warn("消费者队列阻塞中。。。");
                        } catch (InterruptedException e) {
                            // should not be interrupted
                        }
                    }
                }
            });
            while (true) {
                consumptionExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        log.info("消费者消费{}，队列还剩余{}个未消费", r.get(), count);
                    }
                });
            }
        }).start();
    }
}
